package com.viknix.threadpool.manager.config.service.impl;

import com.viknix.threadpool.manager.common.model.PoolParameterInfo;
import com.viknix.threadpool.manager.common.util.Md5Util;
import com.viknix.threadpool.manager.server.dependence.exception.ServerProcessException;
import com.viknix.threadpool.manager.config.executor.ConfigExecutor;
import com.viknix.threadpool.manager.config.model.CacheItem;
import com.viknix.threadpool.manager.server.dependence.monitor.NotifyCenter;
import com.viknix.threadpool.manager.server.dependence.monitor.event.Event;
import com.viknix.threadpool.manager.server.dependence.monitor.event.LocalDataChangeEvent;
import com.viknix.threadpool.manager.server.dependence.monitor.listeners.Listener;
import com.viknix.threadpool.manager.config.service.ConfigService;
import com.viknix.threadpool.manager.config.service.LongPollingService;
import com.viknix.threadpool.manager.config.util.HttpUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.servlet.AsyncContext;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import static com.viknix.threadpool.manager.common.constant.Constants.*;

/**
 * @Author: Dongqi
 * @Date: 2021/10/29 16:37
 * @Version 1.0
 * @Description: LongPollingServiceImpl
 */
@Slf4j
@Service
public class LongPollingServiceImpl implements LongPollingService {

    private static final ConcurrentHashMap<String, CacheItem> CACHE = new ConcurrentHashMap();

    private Map<String, Long> retainIps = new ConcurrentHashMap();

    /**
     * 当前正在执行长轮询的集合
     */
    final Queue<ClientLongPolling> allSubs;

    @Autowired
    private ConfigService configService;

    public LongPollingServiceImpl() {

        allSubs = new ConcurrentLinkedQueue();
        // 定时打印正在执行长轮询的客户端数量
        ConfigExecutor.scheduleLongPolling(new StatTask(), 0L, 30L, TimeUnit.SECONDS);

        // 注册事件多播器
        NotifyCenter.registerToPublisher(LocalDataChangeEvent.class, NotifyCenter.ringBufferSize);

        // 注册事件监听器
        NotifyCenter.registerSubscriber(new Listener() {

            @Override
            public void onEvent(Event event) {
                if (event instanceof LocalDataChangeEvent) {
                    LocalDataChangeEvent dataChangeEvent = (LocalDataChangeEvent) event;
                    ConfigExecutor.executeLongPolling(new DataChangeTask(dataChangeEvent.getGroupKey()));
                }
            }

            @Override
            public Class<? extends Event> subscribeType() {
                return LocalDataChangeEvent.class;
            }
        });
    }

    @Override
    public void addLongPollingClient(HttpServletRequest req, HttpServletResponse rsp, Map<String, String> clientMd5Map, int probeRequestSize) {

        // 获取长轮询的时间
        String str = req.getHeader(LONG_PULLING_TIMEOUT);
        String noHangUpFlag = req.getHeader(LONG_PULLING_TIMEOUT_NO_HANGUP);
        // 客户端会等待29.5s执行
        long timeout = Long.parseLong(str) - 500;
        // 获取客户端需要更新的配置
        List<String> changedGroups = compareMd5(clientMd5Map);
        // 如果有需要更新的配置
        if (changedGroups.size() > 0) {
            // 将所有需要更新的threadPoolId, projectId, namespace拼接起来返回
            HttpUtil.generateResponse(rsp, changedGroups);
            return;
        } else if (noHangUpFlag != null && noHangUpFlag.equalsIgnoreCase(TRUE_STR)) {
            // 如果客户端有新增的长轮询配置，不需要挂起当前服务
            log.info("New initializing cacheData added in.");
            return;
        }
        // 执行长轮询
        String ip = HttpUtil.getRemoteIp(req);
        // 通过req开启异步上下文，将req和resp绑定到上下文中
        final AsyncContext asyncContext = req.startAsync();
        asyncContext.setTimeout(0L);

        // 执行延时任务在29.5s后
        ConfigExecutor.executeLongPolling(new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout));
    }

    /**
     * 判断客户端线程池配置是否需要更新
     *
     * @param clientMd5Map threadPoolId 和 projectId 和 namespace 拼接起来 作为key，线程池配置的 md5 作为 value
     * @return 需要更新的线程池配置：threadPoolId + projectId + namespace
     */
    private List<String> compareMd5(Map<String, String> clientMd5Map) {
        List<String> changedGroupKeys = new ArrayList();
        clientMd5Map.forEach((groupKey, md5) -> {
            // 判断客户端线程池配置是否需要更新
            boolean isUpdateData = isUpdateData(groupKey, md5);
            if (!isUpdateData) {
                // 将key加入到 changedGroupKeys 集合中
                changedGroupKeys.add(groupKey);
            }
        });
        return changedGroupKeys;
    }

    /**
     * 判断客户端线程池配置是否需要更新
     * 先从本地缓存中获取md5，如果没有的话再从数据库查询配置信息
     *
     * @param groupKey threadPoolId + projectId + namespace
     * @param md5      该线程池配置的 md5
     * @return 该客户端线程池配置是否需要更新：true 需要更新；false 不需要更新
     */
    private boolean isUpdateData(String groupKey, String md5) {
        // 从缓存当中找md5的值，没有的话从数据库查询
        String contentMd5 = getContentMd5(groupKey);
        return Objects.equals(contentMd5, md5);
    }

    /**
     * 从缓存当中找到线程池配置的md5值，没有的话从数据库查询
     *
     * @param groupKey threadPoolId + projectId + namespace
     * @return 该线程池配置的 md5
     */
    private String getContentMd5(String groupKey) {

        // 从 本地缓存 cache 当中获取 CacheItem
        CacheItem cacheItem = CACHE.get(groupKey);
        if (cacheItem != null) {
            return cacheItem.getMd5();
        }

        String[] split = groupKey.split("\\+");
        // 根据 threadPoolId、projectId、namespace 从数据库查询 config
        PoolParameterInfo parameterInfo = configService.getParameterInfo(split[0], split[1], split[2]);
        if (parameterInfo != null && StringUtils.isNotBlank(parameterInfo.getThreadPoolId())) {
            // 获取配置的md5值
            String md5 = Md5Util.getTpContentMd5(parameterInfo);
            // 重新构建一个 CacheItem 放到缓存当中
            cacheItem = new CacheItem(groupKey, md5);
            CACHE.put(groupKey, cacheItem);
        } else {
            throw new ServerProcessException("未查询到对应的线程池配置！");
        }
        // 返回当前的md5
        return (cacheItem != null) ? cacheItem.getMd5() : BLANK;
    }

    public Map<String, Long> getRetainIps() {
        return retainIps;
    }

    class ClientLongPolling implements Runnable {

        final AsyncContext asyncContext;

        // key：threadPoolId + projectId + namespace
        // value：md5
        final Map<String, String> clientMd5Map;

        final long createTime;

        final String ip;

        final int probeRequestSize;

        final long timeoutTime;

        Future<?> asyncTimeoutFuture;

        public ClientLongPolling(AsyncContext asyncContext, Map<String, String> clientMd5Map, String ip, int probeRequestSize, long timeout) {

            this.asyncContext = asyncContext;
            // key：threadPoolId + projectId + namespace
            // value：md5
            this.clientMd5Map = clientMd5Map;
            this.ip = ip;
            this.probeRequestSize = probeRequestSize;
            this.timeoutTime = timeout;
            this.createTime = System.currentTimeMillis();
        }

        @Override
        public void run() {

            asyncTimeoutFuture = ConfigExecutor.scheduleLongPolling(() -> {
                try {
                    getRetainIps().put(ClientLongPolling.this.ip, System.currentTimeMillis());
                    // 将当前对象从当前正在执行长轮询的集合集合当中移除
                    allSubs.remove(ClientLongPolling.this);
                    // 响应消息
                    sendResponse(null);
                } catch (Exception e) {
                    log.error("Long polling error :: {}", e.getMessage(), e);
                }
            }, timeoutTime, TimeUnit.MILLISECONDS);

            // 将当前对象加入到当前正在执行长轮询的集合中
            allSubs.add(this);
        }

        private void sendResponse(List<String> changedGroups) {

            // 取消超时任务
            // Cancel time out task.
            if (null != asyncTimeoutFuture) {
                asyncTimeoutFuture.cancel(false);
            }
            HttpUtil.generateAsyncResponse(asyncContext, changedGroups);
        }
    }

    class StatTask implements Runnable {

        @Override
        public void run() {
            log.info("Dynamic Thread Pool Long pulling client count :: {}", allSubs.size());
        }
    }

    class DataChangeTask implements Runnable {

        final String groupKey;

        DataChangeTask(String groupKey) {
            this.groupKey = groupKey;
        }

        @Override
        public void run() {
            try {
                for (Iterator<ClientLongPolling> iter = allSubs.iterator(); iter.hasNext(); ) {
                    // 循环当前正在执行长轮询的客户端
                    ClientLongPolling clientSub = iter.next();
                    // 如果该客户端的配置中存在当前 groupKey
                    if (clientSub.clientMd5Map.containsKey(groupKey)) {
                        // 更新客户端的刷新时间
                        getRetainIps().put(clientSub.ip, System.currentTimeMillis());
                        updateMd5(groupKey, configService.getContentMd5(groupKey), System.currentTimeMillis());
                        // 从 allSubs 中移除该客户端
                        iter.remove();
                        // 响应客户端
                        clientSub.sendResponse(Arrays.asList(groupKey));
                    }
                }
            } catch (Exception ex) {
                log.error("Data change error :: {}", ex.getMessage(), ex);
            }
        }
    }

    private static void updateMd5(String groupKey, String md5, long lastModifiedTs) {
        if (StringUtils.isBlank(md5)) {
            removeCache(groupKey);
            return;
        }
        CacheItem cache = makeSure(groupKey);
        if (cache.getMd5() == null || !cache.getMd5().equals(md5)) {
            cache.setMd5(md5);
            cache.setLastModifiedTs(lastModifiedTs);
            NotifyCenter.publishEvent(new LocalDataChangeEvent(groupKey));
        }
    }

    private static CacheItem makeSure(final String groupKey) {
        CacheItem item = CACHE.get(groupKey);
        if (null != item) {
            return item;
        }
        CacheItem tmp = new CacheItem(groupKey);
        item = CACHE.putIfAbsent(groupKey, tmp);
        return (null == item) ? tmp : item;
    }

    private static boolean removeCache(final String groupKey) {
        CacheItem groupItem = CACHE.get(groupKey);
        int lockResult = (null == groupItem) ? 0 : (groupItem.rwLock.tryWriteLock() ? 1 : -1);
        if (lockResult < 0) {
            log.warn("[dump-error] write lock failed. {}", groupKey);
            return false;
        }
        try {
            CACHE.remove(groupKey);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        } finally {
            if (Objects.nonNull(groupItem)) {
                groupItem.rwLock.releaseWriteLock();
            }
        }
        return true;
    }

}
